[FLINK-35157][runtime] Sources with watermark alignment get stuck once some subtasks finish #24757

elon-X · 2024-05-06T14:52:48Z

What is the purpose of the change

Sources with watermark alignment get stuck once some subtasks finish, this PR solves this problem.

Brief change log

while some subtasks have been finished, the SourceOperator send Long.MAX_VALUE to SourceCoordinator, and SourceCoordinator checks whether subtasks have been finished before sending the event.

Verifying this change

This change added tests and can be verified as follows:

org.apache.flink.streaming.api.operators.SourceOperatorAlignmentTest::testWatermarkAlignmentWhileSubtaskFinished()
org.apache.flink.test.streaming.api.datastream.WatermarkAlignmentITCase::testTaskFinishedWithWatermarkAlignmentExecution()

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): no
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? no

flinkbot · 2024-05-06T14:55:36Z

CI report:

8ac3d8a Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

elon-X · 2024-05-07T02:31:15Z

hi, @1996fanrui would you mind reviewing this for me when you have a moment? Thank you very much!

1996fanrui

Hi @elon-X , I'm wondering if "NoMoreSplitsEvent" would be appropriate to check if a subtask is finished.

The Split can be discovered dynamically. IIRC, kafka source subtask won't be FINISHED even if it doesn't have split for Flink Streaming Job. (This subtask may be assigned split after adding new kafka partition)

I prefer Gyula proposed solution: The solution could be to send out a max watermark event once the sources finish or to exclude them from the source coordinator.

We need to find a proper code place(source is definitely finished) to send MaxTimestamp.

I guess

flink/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java

Line 452 in 046872c

return DataInputStatus.END_OF_INPUT;

may be a suitable place.

flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinator.java

.../apache/flink/test/streaming/api/datastream/SubTaskFinishedWithWatermarkAlignmentITCase.java

elon-X · 2024-06-05T05:46:36Z

@1996fanrui I've made some changes based on your suggestions. Please review the changes when you have a chance and let me know if there are any further improvements needed. Thanks!

1996fanrui

Hi @elon-X , thanks for the update!

The production code looks good to me, I left some comments for test code, please take a look in your free time, thanks~

flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinator.java

...s/src/test/java/org/apache/flink/test/streaming/api/datastream/WatermarkAlignmentITCase.java

...rc/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorAlignmentTest.java

1996fanrui · 2024-06-05T06:29:35Z

...rc/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorAlignmentTest.java

@@ -377,4 +424,9 @@ private void assertLatestWatermarkAlignmentEvent(int subtask, long expectedWater
        assertThat(events.get(events.size() - 1))
                .isEqualTo(new WatermarkAlignmentEvent(expectedWatermark));
    }
+
+    private void assertHasNoMoreSplits(int subtask, boolean expected) {


method name is assertHasNoMoreSplits, but it triggers signalNoMoreSplits as well.

I think sourceCoordinator.getContext().signalNoMoreSplits(subtask); should be moved out of this method.

Also, assertHasNoMoreSplits is only called once, and the method only has one line. I'm not sure whether we need extract it as a separate method. Caller can call assertThat(sourceCoordinator.getContext().hasNoMoreSplits(subtask)).isEqualTo(expected); directly.

1996fanrui · 2024-06-05T06:34:57Z

...rc/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorAlignmentTest.java

+        assertLatestWatermarkAlignmentEvent(subtask1, 1042);
+
+        // mock noMoreSplits event
+        assertHasNoMoreSplits(subtask0, true);


I'm not sure what testWatermarkAlignmentWhileSubtaskFinished wants to test?

IIUC, the test still works even if we don't call assertHasNoMoreSplits(subtask0, true); (remove this line) here, right?

My intention is to add a test case to simulate the sending of the noMoreSplits event, indicating that the SourceCoordinator is working correctly. If we have WatermarkAlignmentITCase, perhaps we can remove testWatermarkAlignmentWhileSubtaskFinished. What do you think?

My intention is to add a test case to simulate the sending of the noMoreSplits event

Actually, this test is testing the watermark result when the watermark of one subtask is Long.Max, right?

I think this test can be removed.

Updated, thanks for your patience.

1996fanrui

Thanks @elon-X for the update!

LGTM assuming the CI is green.

1996fanrui · 2024-06-13T06:46:41Z

@elon-X The CI fails, could you rebase the master branch first? We can follow the CI after rebaseing.

…e some subtasks finish

…e some subtasks finish (apache#24757)

…e some subtasks finish (#24757)

…e some subtasks finish (apache#24757)

flinkbot added the component=Runtime/Coordination label May 6, 2024

1996fanrui self-requested a review May 10, 2024 08:10

1996fanrui self-assigned this May 10, 2024

1996fanrui reviewed May 10, 2024

View reviewed changes

flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinator.java Show resolved Hide resolved

1996fanrui reviewed Jun 4, 2024

View reviewed changes

1996fanrui reviewed Jun 5, 2024

View reviewed changes

1996fanrui approved these changes Jun 13, 2024

View reviewed changes

elon-X added 5 commits June 13, 2024 22:26

[FLINK-35157][runtime] Sources with watermark alignment get stuck onc…

71be209

…e some subtasks finish

spotless-check

1d52c20

adjusted the code layout and added an ITCase

422d16d

fixed watermarkAlignmentITCase

460ffd0

fixed test code

8ac3d8a

elon-X force-pushed the watermark-alignment branch from ad0d3c8 to 8ac3d8a Compare June 13, 2024 14:26

1996fanrui merged commit 93b9bdf into apache:master Jun 14, 2024

elon-X added a commit to elon-X/flink that referenced this pull request Jun 16, 2024

[FLINK-35157][runtime] Sources with watermark alignment get stuck onc…

8899d14

…e some subtasks finish (apache#24757)

elon-X mentioned this pull request Jun 16, 2024

[FLINK-35157][runtime] Sources with watermark alignment get stuck once some subtasks finish #24942

Merged

elon-X added a commit to elon-X/flink that referenced this pull request Jun 16, 2024

[FLINK-35157][runtime] Sources with watermark alignment get stuck onc…

52f30d7

…e some subtasks finish (apache#24757)

elon-X added a commit to elon-X/flink that referenced this pull request Jun 16, 2024

[FLINK-35157][runtime] Sources with watermark alignment get stuck onc…

d5818ed

…e some subtasks finish (apache#24757)

1996fanrui pushed a commit that referenced this pull request Jun 18, 2024

[FLINK-35157][runtime] Sources with watermark alignment get stuck onc…

dc89ffb

…e some subtasks finish (#24757)

1996fanrui pushed a commit that referenced this pull request Jun 18, 2024

[FLINK-35157][runtime] Sources with watermark alignment get stuck onc…

72fac4c

…e some subtasks finish (#24757)

1996fanrui pushed a commit that referenced this pull request Jun 18, 2024

[FLINK-35157][runtime] Sources with watermark alignment get stuck onc…

40cb39e

…e some subtasks finish (#24757)

superdiaodiao pushed a commit to superdiaodiao/flink that referenced this pull request Jul 4, 2024

[FLINK-35157][runtime] Sources with watermark alignment get stuck onc…

dadf1a8

…e some subtasks finish (apache#24757)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-35157][runtime] Sources with watermark alignment get stuck once some subtasks finish #24757

[FLINK-35157][runtime] Sources with watermark alignment get stuck once some subtasks finish #24757

elon-X commented May 6, 2024 •

edited

Loading

flinkbot commented May 6, 2024 •

edited

Loading

elon-X commented May 7, 2024

1996fanrui left a comment

elon-X commented Jun 5, 2024

1996fanrui left a comment

1996fanrui Jun 5, 2024

1996fanrui Jun 5, 2024

elon-X Jun 12, 2024

1996fanrui Jun 13, 2024

elon-X Jun 13, 2024

1996fanrui left a comment

1996fanrui commented Jun 13, 2024

[FLINK-35157][runtime] Sources with watermark alignment get stuck once some subtasks finish #24757

[FLINK-35157][runtime] Sources with watermark alignment get stuck once some subtasks finish #24757

Conversation

elon-X commented May 6, 2024 • edited Loading

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented May 6, 2024 • edited Loading

CI report:

elon-X commented May 7, 2024

1996fanrui left a comment

Choose a reason for hiding this comment

elon-X commented Jun 5, 2024

1996fanrui left a comment

Choose a reason for hiding this comment

1996fanrui Jun 5, 2024

Choose a reason for hiding this comment

1996fanrui Jun 5, 2024

Choose a reason for hiding this comment

elon-X Jun 12, 2024

Choose a reason for hiding this comment

1996fanrui Jun 13, 2024

Choose a reason for hiding this comment

elon-X Jun 13, 2024

Choose a reason for hiding this comment

1996fanrui left a comment

Choose a reason for hiding this comment

1996fanrui commented Jun 13, 2024

elon-X commented May 6, 2024 •

edited

Loading

flinkbot commented May 6, 2024 •

edited

Loading